package com.wang.helloworld.storm;

import java.util.Scanner;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;

import com.wang.helloworld.storm.bolt.CommandMessageBolt;
import com.wang.helloworld.storm.bolt.MessageDeliveryBolt;
import com.wang.helloworld.storm.bolt.OtherMessageBolt;
import com.wang.helloworld.storm.spout.KafkaConsumerSpout;

/**
 * Hello world!
 *
 */
public class HelloStorm 
{
	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
		HelloStorm storm = new HelloStorm();
		storm.run();
	}

	public HelloStorm() throws AlreadyAliveException, InvalidTopologyException {
		Config conf = new Config();
		conf.setNumWorkers(2);
		conf.setMaxSpoutPending(5000);
		conf.put(Config.TOPOLOGY_WORKERS, 4);
		StormSubmitter.submitTopology("kafkaConsumerTopology", conf, buildTopoloty());
	}

    public StormTopology buildTopoloty()
    {
    	TopologyBuilder builder = new TopologyBuilder();
		builder = new TopologyBuilder();
		
		builder.setSpout("subscribeMessage", new KafkaConsumerSpout(), 2);

		builder.setBolt("deliveryMessage", new MessageDeliveryBolt(), 2).shuffleGrouping("subscribeMessage");
		
		builder.setBolt("processMesage1", new CommandMessageBolt(), 1).shuffleGrouping("deliveryMessage", "processCommandMessage");
		builder.setBolt("processMesage2", new OtherMessageBolt(), 1).shuffleGrouping("deliveryMessage", "processOtherMessage");
		
		return builder.createTopology();
    }

	public void run() {
		Scanner console = new Scanner(System.in);
		do {
			System.out.print("> ");
			String command = console.nextLine();
			if (command.equals("exit")) {
				return;
			}
		} while (true);
	}
}
